package com.haozhen.streamsink;

import com.haozhen.entity.User;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.net.UnknownServiceException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;

/**
 * @author haozhen
 * @email haozh@ync1.com
 * @date 2021/2/26  0:42
 */
public class MySinkToMysql {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        ArrayList<User> users = new ArrayList<>();

        users.add(new User("xt","male","北京"));
        users.add(new User("gxy","male","北京"));
        users.add(new User("qqh","male","北京"));
        DataStreamSource<User> userDataStreamSource = env.fromCollection(users);

        String url = "jdbc:mysql://localhost:3306/test1?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
        String user="root";
        String password = "root";

        userDataStreamSource.addSink(new RichSinkFunction<User>() {

            Connection connection = null;

            PreparedStatement preparedStatement = null;

            @Override
            public void invoke(User value, Context context) throws Exception {
                preparedStatement.setString(1,value.getUsername());
                preparedStatement.setString(2,value.getSex());
                preparedStatement.setString(3,value.getAddress());
                int i = preparedStatement.executeUpdate();

            }

            /**
             *  初始化 mysql 的连接
             * @param parameters
             * @throws Exception
             */
            @Override
            public void open(Configuration parameters) throws Exception {

                 connection = DriverManager.getConnection(url, user, password);

                String sql = "insert into sys_user (username,sex,address) values(?,?,?)";

                preparedStatement = connection.prepareStatement(sql);
            }

            @Override
            public void close() throws Exception {
               if(preparedStatement!=null){
                   preparedStatement.close();
               }
               if(connection!=null){
                   connection.close();
               }
            }
        });
        env.execute();
    }
}
